package com.example.reactor;

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;

@Slf4j
public class Practice3 {

    public static void main(String[] args) {
        // 结果回调在弹性线程池的线程上
        Mono.just("data")
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String data) {
                        // 当前线程：主线程
                        log.info(data);
                        return data.toUpperCase();
                    }
                })
                .publishOn(Schedulers.parallel()) // 切换到并行线程池
                .flatMap(new Function<String, Mono<? extends String>>() {
                    @Override
                    public Mono<? extends String> apply(String data) {
                        // 当前线程：并行线程池中的线程
                        log.info(data);
                        return Mono.just(data + " processed");
                    }
                })
                .subscribeOn(Schedulers.boundedElastic()) // 订阅数据的操作会在弹性线程池上执行
                .subscribe(log::info);

        List<Integer> list = new ArrayList<>();
        list.add(1);
        list.add(2);
        list.add(3);
        Flux.fromIterable(list)
                .collectList()
                .map(
                        new Function<List<Integer>, List<Integer>>() {
                            @Override
                            public List<Integer> apply(List<Integer> integers) {
                                log.info(integers.toString());
                                return integers;
                            }
                        }
                ).subscribe();


    }
}
